#pragma once
#include <muduo/net/TcpServer.h>
#include <muduo/net/TcpClient.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/EventLoopThread.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/base/CountDownLatch.h>
#include <muduo/net/Buffer.h>
#include <unordered_map>
#include <mutex>
#include "detail.hpp"
#include "fields.hpp"
#include "abstract.hpp"
#include "Log.hpp"
#include "message.hpp"

namespace yjz_rpc
{
    using namespace LogModule;
    
    class MuduoBuffer : public BaseBuffer
    {
    public:
        using ptr = std::shared_ptr<MuduoBuffer>;
        
        MuduoBuffer(muduo::net::Buffer *buf) 
            : _buf(buf)
        {}

        virtual size_t readableSize() override
        {
            return _buf->readableBytes();
        }

        virtual int32_t peekInt32() override
        {
            return _buf->peekInt32();
        }                    

        virtual void retrieveInt32() override
        {
            _buf->retrieveInt32();
        }         

        virtual int32_t readInt32() override    
        {
            return _buf->readInt32();
        }

        virtual std::string retrieveAsString(size_t len) override    
        {
            return _buf->retrieveAsString(len);
        }

    private:
        muduo::net::Buffer* _buf;
    };

    class BufferFactory
    {
    public:
        template<class ...Args>
        static BaseBuffer::ptr create(Args&& ...args)
        {
            return std::make_shared<MuduoBuffer>(std::forward<Args>(args)...);
        }
    };

    class LVProtocol : public BaseProtocol
    {
        // | len | mtype | idlenth | id | body |
    public:
        using ptr = std::shared_ptr<LVProtocol>;

        virtual bool canProcessed(const BaseBuffer::ptr& buf) override
        {
            if (buf->readableSize() < lenFieldsLength) 
            {
                return false;
            }
            // 判断缓冲区中的数据是否足够处理一条完整的消息
            if (buf->readableSize() < buf->peekInt32() + lenFieldsLength)
            {
                return false;
            }
            return true;
        }

        virtual bool onMessage(const BaseBuffer::ptr& buf, BaseMessage::ptr& msg) override
        {
            // 调用onMessage默认当前缓冲区中的数据满足处理一条完整消息
            int32_t total_len = buf->readInt32(); // 读取总长度
            MType mtype = (MType)buf->readInt32(); // 读取消息类型
            int32_t idlen = buf->readInt32(); // 读取id长度
            std::string id = buf->retrieveAsString(idlen);
            int32_t body_len = total_len - mtypeFieldsLength - idFieldsLength - idlen;
            std::string body = buf->retrieveAsString(body_len);
            msg = MessageFactory::create(mtype);
            if (msg.get() == nullptr)
            {
                LOG(LogLevel::ERROR) << "消息类型错误，构造消息对象失败！";
                return false;
            }
            bool res = msg->unserialize(body);
            if (res == false)
            {
                LOG(LogLevel::ERROR) << "消息正文反序列化失败！";
                return false;
            }
            msg->setId(id);
            msg->setMType(mtype);
            return true;
        }

        virtual std::string serialize(const BaseMessage::ptr& msg) override
        {
            std::string body = msg->serialize();
            std::string id = msg->getId();
            auto mtype = htonl((int32_t)msg->getMType());
            int32_t idlen = htonl(id.size());
            int32_t h_total_len = mtypeFieldsLength + idFieldsLength + id.size() + body.size();
            int32_t n_total_len = htonl(h_total_len);
            std::string result;
            result.reserve(h_total_len);
            result.append((char*)&n_total_len, lenFieldsLength);
            result.append((char*)&mtype, mtypeFieldsLength);
            result.append((char*)&idlen, idFieldsLength);
            result.append(id);
            result.append(body);
            return result;
        }
    private:
        const size_t lenFieldsLength = 4;
        const size_t mtypeFieldsLength = 4;
        const size_t idFieldsLength = 4;
    };

    class ProtocolFactory
    {
    public:
        template<class ...Args>
        static BaseProtocol::ptr create(Args&& ...args)
        {
            return std::make_shared<LVProtocol>(std::forward<Args>(args)...);
        }
    };

    class MuduoConnection : public BaseConnection
    {
    public:
        using ptr = std::shared_ptr<MuduoConnection>;

        MuduoConnection(const BaseProtocol::ptr &protocol, const muduo::net::TcpConnectionPtr &conn)
            : _protocol(protocol), _conn(conn)
        {}

        virtual void send(const BaseMessage::ptr& msg) override
        {
            std::string body = _protocol->serialize(msg);
            _conn->send(body);
        }

        virtual void shutdown() override
        {
            _conn->shutdown();
        }

        virtual bool connected() override
        {
            return _conn->connected();
        }
    private:
        BaseProtocol::ptr _protocol;
        muduo::net::TcpConnectionPtr _conn;
    };

    class ConnectionFactory
    {
    public:
        template<class ...Args>
        static BaseConnection::ptr create(Args&& ...args)
        {
            return std::make_shared<MuduoConnection>(std::forward<Args>(args)...);
        }
    };

    class MuduoServer : public BaseServer
    {
    public:
        using ptr = std::shared_ptr<BaseServer>;
        
        MuduoServer(int port)
            : _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port),
            "MuduoServer", muduo::net::TcpServer::kReusePort), 
            _protocol(ProtocolFactory::create())
        {}

        virtual void start() override
        {
            // 设置连接事件的回调
            _server.setConnectionCallback(std::bind(&MuduoServer::onConnection, this, std::placeholders::_1));
            // 设置连接消息的回调
            _server.setMessageCallback(std::bind(&MuduoServer::onMessage, this, std::placeholders::_1, 
                std::placeholders::_2, std::placeholders::_3));

            _server.start();// 先开始监听
            _baseloop.loop();// 开始死循环事件监控
        }
    private:
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                LOG(LogLevel::DEBUG) << "连接建立！";
                auto muduo_conn = ConnectionFactory::create(_protocol, conn);
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _conns[conn] = muduo_conn;
                }
                if (_connection) _connection(muduo_conn);
            }
            else 
            {
                LOG(LogLevel::WARNING) << "连接断开！";
                BaseConnection::ptr muduo_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if (it == _conns.end())
                    {
                        return;
                    }
                    muduo_conn = it->second;
                    _conns.erase(conn);
                }
                if (_close) _close(muduo_conn);
            }
        }
    
        void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp)
        {
            LOG(LogLevel::DEBUG) << "连接有新数据到来，开始处理！";
            auto base_buf = BufferFactory::create(buf);
            while (1)
            {
                if (_protocol->canProcessed(base_buf) == false)
                {
                    // 缓冲区中数据很多，但因为数据错误导致数据又不足一条完整消息的处理，这种情况下关闭连接
                    if (base_buf->readableSize() > maxDataSize)
                    {
                        conn->shutdown();
                        LOG(LogLevel::ERROR) << "缓冲区中数据多大!";
                        return;
                    }
                    LOG(LogLevel::DEBUG) << "数据量不足！";
                    break;
                }
                BaseMessage::ptr msg;
                bool res = _protocol->onMessage(base_buf, msg);
                if (res == false)
                {
                    conn->shutdown();
                    LOG(LogLevel::ERROR) << "缓冲区中数据错误！";
                    return;
                }
                LOG(LogLevel::DEBUG) << "消息反序列化成功！";
                BaseConnection::ptr base_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if (it == _conns.end())
                    {
                        conn->shutdown();
                        return;
                    }
                    base_conn = it->second;
                }
                if (_message) _message(base_conn, msg);
            }
        }
    private:
        const size_t maxDataSize = (1 << 16);
        BaseProtocol::ptr _protocol;
        muduo::net::EventLoop _baseloop; // 事件循环监控
        muduo::net::TcpServer _server; // 通信连接管理
        std::unordered_map<muduo::net::TcpConnectionPtr, BaseConnection::ptr> _conns;
        std::mutex _mutex;
    };

    class ServerFactory
    {
    public:
        template<class ...Args>
        static BaseServer::ptr create(Args&& ...args)
        {
            return std::make_shared<MuduoServer>(std::forward<Args>(args)...);
        }
    };

    class MuduoClient : public BaseClient
    {
    public:
        using ptr = std::shared_ptr<MuduoClient>;
        
        MuduoClient(const std::string &sip, int port)
            : _protocol(ProtocolFactory::create())
            , _baseloop(_loopthread.startLoop())
            , _downlatch(1) // 初始化计数器为1，为0时唤醒
            , _client(_baseloop, muduo::net::InetAddress(sip, port), "MuduoClient")
        {}

        virtual void connect() override
        {
            // 设置连接事件的回调
            _client.setConnectionCallback(std::bind(&MuduoClient::onConnection, this, std::placeholders::_1));
            // 设置连接消息的回调
            _client.setMessageCallback(std::bind(&MuduoClient::onMessage, this, std::placeholders::_1, 
                                        std::placeholders::_2, std::placeholders::_3));

            // 连接服务器
            _client.connect();
            _downlatch.wait();
        }

        virtual void shutdown() override
        {
            _client.disconnect();
        }

        virtual bool send(const BaseMessage::ptr& msg) override
        {
            if (connected() == false)
            {
                LOG(LogLevel::ERROR) << "连接已断开！";
                return false;
            }
            _conn->send(msg);
            return true;
        }

        virtual BaseConnection::ptr getConnect() override
        {
            return _conn;
        }

        virtual bool connected() override
        {
            return (_conn && _conn->connected());
        }
    private:
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                LOG(LogLevel::WARNING) << "连接建立！";
                _downlatch.countDown();// --, 计数为0时唤醒阻塞
                _conn = ConnectionFactory::create(_protocol, conn);
            }
            else 
            {
                LOG(LogLevel::WARNING) << "连接断开！";
                _conn.reset();
            }
        }
    
        void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp)
        {
            LOG(LogLevel::DEBUG) << "连接有新数据到来，开始处理！";
            auto base_buf = BufferFactory::create(buf);
            while (1)
            {
                if (_protocol->canProcessed(base_buf) == false)
                {
                    // 缓冲区中数据很多，但因为数据错误导致数据又不足一条完整消息的处理，这种情况下关闭连接
                    if (base_buf->readableSize() > maxDataSize)
                    {
                        conn->shutdown();
                        LOG(LogLevel::ERROR) << "缓冲区中数据多大!";
                        return;
                    }
                    LOG(LogLevel::DEBUG) << "数据量不足！";
                    break;
                }
                BaseMessage::ptr msg;
                bool res = _protocol->onMessage(base_buf, msg);
                if (res == false)
                {
                    conn->shutdown();
                    LOG(LogLevel::ERROR) << "缓冲区中数据错误！";
                    return;
                }
                LOG(LogLevel::DEBUG) << "缓冲区中数据解析完毕，调用回调函数处理！";
                if (_message) _message(_conn, msg);
            }
        }
    private:
        const size_t maxDataSize = (1 << 16);
        BaseProtocol::ptr _protocol;
        BaseConnection::ptr _conn;
        muduo::CountDownLatch _downlatch;
        muduo::net::EventLoopThread _loopthread;
        muduo::net::EventLoop *_baseloop;
        muduo::net::TcpClient _client;
    };

    class ClientFactory
    {
    public:
        template<class ...Args>
        static BaseClient::ptr create(Args&& ...args)
        {
            return std::make_shared<MuduoClient>(std::forward<Args>(args)...);
        }
    };
}